package com.cellulam.framework.mq.listener;

import org.apache.commons.collections.CollectionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.boot.CommandLineRunner;
import org.springframework.stereotype.Component;

import java.util.List;

/**
 * mq消费者初始化
 */
@Component
public class ConsumerInitializer implements CommandLineRunner {

    private static final Logger logger = LoggerFactory.getLogger(ConsumerInitializer.class);

    @Override
    public void run(String... args) throws Exception {
        List<MqConsumeInfo> consumers = MqConsumerHolder.holder.getMqListeners();

        logger.info("consumer initializer and consume size is {}", consumers.size());
        if (CollectionUtils.isEmpty(consumers)) {
            return;
        }

        for (MqConsumeInfo consumer : consumers) {
            MqListener subscriber = consumer.getSubscriber();
            subscriber.initConsumer(consumer.getConsumerConfig());

            subscriber.subscribe(consumer.getGroupId(),
                    consumer.getConsumerConfig().getTopic(),
                    this.convertToSubExpression(consumer.getConsumerConfig().getTags()));
            subscriber.start();
        }
        logger.info("done init mq consumer");
    }

    private String convertToSubExpression(List<String> tags) {
        if (CollectionUtils.isEmpty(tags)) {
            return "*";
        }

        return String.join("||", tags);
    }

}
